/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.transfer;

import com.dtdream.emr.transmission.exception.TransmissionException;
import com.dtdream.emr.transmission.io.RequestDecoder;
import com.dtdream.emr.transmission.io.RequestEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_NETWORK;
import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_SERVER;

public class DataChannel<T> {
    private static final Logger LOG = LoggerFactory.getLogger(DataChannel.class);

    private Channel channel;
    private Bootstrap b;
    private String host;
    private int port;
    private static final int SEND_WAIT_INTERVAL = 500;

    /**
     * @param host, server host
     * @param port, server port
     */
    public DataChannel(String host, int port, EventLoopGroup workerGroup, NettyConfig nettyConfig) {
        this.b = new Bootstrap();
        b.group(workerGroup);
        b.channel(NioSocketChannel.class);
        b.option(ChannelOption.SO_KEEPALIVE, true);
        b.option(ChannelOption.TCP_NODELAY, true);
        b.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(nettyConfig.getWaterMark() / 2,
                nettyConfig.getWaterMark()));
        b.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new RequestDecoder(), new RequestEncoder(), new ClientHandler());
            }
        });
        this.host = host;
        this.port = port;
    }

    public void init() throws TransmissionException {
        try {
            ChannelFuture channelFuture = b.connect(host, port).sync();
            channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
                public void operationComplete(Future<? super Void> future) throws Exception {
                    if (!future.isSuccess()) {
                        LOG.error("Data transfer error", future.cause());
                    }
                }
            });
            this.channel = channelFuture.channel();
        } catch (Exception e) {
            LOG.error("Init chancel failed!", e);
            throw new TransmissionException(EXCEPTION_SDK_NETWORK, e);
        }
    }

    public void send(T t) throws TransmissionException {
        try {
            while (!channel.isWritable()) {
                if (!channel.isActive()) {
                    throw new TransmissionException(EXCEPTION_SDK_SERVER, "channel was closed abnormally");
                }
                synchronized (this) {
                    this.wait(SEND_WAIT_INTERVAL);
                }
            }
        } catch (InterruptedException e) {
            LOG.error("send wait thread interrupted", e);
        }
        channel.writeAndFlush(t);
    }

    public void close() {
        try {
            if (null != channel) {
                channel.close().sync();
            }
        } catch (InterruptedException e) {
            LOG.error("channel close error", e);
        }
    }
}